package com.yeming.flink.practice.transformation

import java.text.SimpleDateFormat
import java.util.Date

import com.yeming.flink.practice.source.StationLog
import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.streaming.api.scala._

object TranMapFunctionClass {

  // 计算出每个通话的起始时间与结束时间
  //数据源来源于本地
  def main(args: Array[String]): Unit = {

    //加载环境
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //读取数据源
    val fileFile: String = getClass.getResource("/station.log").getPath
    val stream: DataStream[String] = streamEnv.readTextFile(fileFile)
    //数据计算，计算通话时间，开始时间和结束时间
    val streamLog: DataStream[StationLog] = stream.map(line => {
      val arr: Array[String] = line.split(",")
      new StationLog(arr(0).trim, arr(1).trim, arr(2).trim, arr(3).trim, arr(4).trim.toLong, arr(5).trim.toLong)
    })

    val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

    val result: DataStream[String] = streamLog.filter(log => log.callType.equals("success"))
      .map(new MyFunction(format))


    result.print("通话记录")
    streamEnv.execute("通话")

  }

  class MyFunction(format:SimpleDateFormat) extends MapFunction[StationLog,String]{
    override def map(value: StationLog): String = {
      var startTime = value.callTime
      var endTime = value.callTime + value.duration * 1000
      String.format("主叫号码：%s, 被叫号码：%s, 通话开始时间：%s, 通话结束时间：%s, 通话时长：%d",value.callOut, value.callIn,
        format.format(new Date(startTime)), format.format(new Date(endTime)), Long.box(value.duration))
    }
  }
}
